package com.fishleap.util;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidDataSourceFactory;
import com.alibaba.fastjson.JSON;
import com.fishleap.controller.UserController;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @author zlg
 * 封装Curator客户端相关Zookeeper API操作
 * 参考链接: https://blog.csdn.net/wo541075754/article/details/70167841
 */
@Slf4j
public class CuratorUtils {
    // 基础睡眠时间
    private static final int BASE_SLEEP_TIME = 1000;
    // 重大重试次数
    private static final int MAX_RETRIES = 3;
    // server地址
    private static final String CONNECT_STRING = "127.0.0.1:2181";
    // 会话超时时间
    private static final int SESSION_TIMEOUT_MS = 5000;
    // 连接超时时间
    private static final int CONNECTION_TIMEOUT_MS = 30000;
    // 独立命名空间
    public static final String NAMESPACE = "mysql-config";
    // 服务地址,key是serviceName,value是子节点列表
    private static Map<String, List<String>> serviceAddressMap = new ConcurrentHashMap<>();
    // 注册中心
    private static Set<String> registeredPathSet = ConcurrentHashMap.newKeySet();
    // Zookeeper客户端
    public static CuratorFramework zkClient;

    static {
        zkClient = getZkClient();
    }

    /**
     * 1.使用Fluent风格创建Zk客户端
     * @return CuratorFramework Zk客户端
     */
    private static CuratorFramework getZkClient() {
        // 重试策略,重试三次,会增加重试之间的睡眠时间
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_STRING)  // server地址
                .sessionTimeoutMs(SESSION_TIMEOUT_MS)     // 会话超时时间
                .connectionTimeoutMs(CONNECTION_TIMEOUT_MS)  // 连接超时时间
                .retryPolicy(retryPolicy)   // 重试策略
                .namespace(NAMESPACE)  // 独立命名空间
                .build();
        client.start();
        log.info("Zookeeper session established. ");
        System.out.println("Zookeeper session established. ");
        return client;
    }

    /**
     * 2.创建持久化节点。不同于临时节点，持久化节点不会因为客户端断开连接而被删除
     * @param path 创建节点的路径,即名称
     */
    public static void createPersistentNode(String path, String data) {
        try {
            if (registeredPathSet.contains(path) || zkClient.checkExists().forPath(path) != null) {
                log.info("持久化节点已经存在，节点为:[{}]", path);
                System.out.println("持久化节点已经存在，节点为:["+path+"]");
            } else {
                //eg: /zdy-rpc/com.fishleap.service.IUserService/127.0.0.1:8888
                zkClient.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(path,data.getBytes());
                log.info("持久化节点创建成功，节点为:[{}]", path);
                System.out.println("持久化节点创建成功，节点为:["+path+"]");
            }
            // 将创建的节点信息保存到set中
            registeredPathSet.add(path);
        } catch (Exception e) {
            e.getMessage();
        }
    }

    /**
     * 创建临时节点,临时节点会因为客户端断开连接而被删除
     * @param path 创建节点的路径,即名称
     */
    public static void createEphemeralNode(String path) {
        try {
            if (registeredPathSet.contains(path) || zkClient.checkExists().forPath(path) != null) {
                log.info("临时节点已经存在，节点为:[{}]", path);
                System.out.println("临时节点已经存在，节点为:["+path+"]");
            } else {
                //eg: /zdy-rpc/com.fishleap.service.IUserService/127.0.0.1:8888
                zkClient.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(path);
                log.info("临时节点创建成功，节点为:[{}]", path);
                System.out.println("临时节点创建成功，节点为:["+path+"]");
            }
            // 将创建的节点信息保存到set中
            registeredPathSet.add(path);
        } catch (Exception e) {
            e.getMessage();
        }
    }

    /**
     * 3.获取某个字节下的子节点,也就是获取所有提供服务的生产者的地址
     * @param serviceName 服务对象接口名 eg:com.fishleap.service.IUserService
     * @return List<String> 指定字节下的所有子节点
     */
    public static List<String> getChildrenNodes(String serviceName) {
        // 判断map中是否有该serviceName的key
        if (serviceAddressMap.containsKey(serviceName)) {
            return serviceAddressMap.get(serviceName);
        }
        List<String> result = null;
        String servicePath = "/" + serviceName;
        try {
            result = zkClient.getChildren().forPath(servicePath);
            log.info("当前 {} 节点的子节点列表为 {}", servicePath, result);
            System.out.println("当前 "+servicePath+" 节点的子节点列表为 "+result);
            serviceAddressMap.put(serviceName, result);
            // 注册节点子节点监听
            registerWatcher(zkClient, serviceName);
        } catch (Exception e) {
            e.getMessage();
        }
        return result;
    }

    /**
     * 4.获取某一节点的数据信息
     * @param path 节点的路径,不带"/"
     * @return 返回节点数据字符串
     */
    public static String getNodeData(String path) {
        Stat stat = new Stat();
        byte[] bytes = new byte[0];
        try {
            bytes = zkClient.getData().storingStatIn(stat).forPath(path);
            // 注册节点监听
            registerWatcherNodeData(zkClient, path);
        } catch (Exception e) {
            e.getMessage();
        }
        return new String(new String(bytes));
    }

    /**
     * 修改指定节点的数据
     * @param path 节点路径
     * @param data 节点数据
     */
    public static void setNodeData(String path, String data) {
        try {
            Stat stat = zkClient.setData().withVersion(-1).forPath(path, data.getBytes());
            if (stat != null) {
                System.out.println("修改Zookeeper上数据库配置信息成功!");
            }
        } catch (Exception e) {
            e.getMessage();
        }
    }

    /**
     * 5.清空注册中心的数据
     */
    public static void clearRegistry() {
        // 遍历注册中心 path 集合
        registeredPathSet.stream().parallel().forEach(path -> {
            try {
                zkClient.delete().forPath(path);
            } catch (Exception e) {
                e.getMessage();
            }
        });
        log.info("服务端（Provider）所有注册的服务都被清空:[{}]", registeredPathSet.toString());
        System.out.println("服务端（Provider）所有注册的服务都被清空:["+registeredPathSet.toString()+"]");
    }


    /**
     * 对指定节点本身内容进行监听
     * @param zkClient zk客户端
     * @param path 节点路径
     */
    private static void registerWatcherNodeData(CuratorFramework zkClient, String path) {
        final NodeCache nodeCache = new NodeCache(zkClient, path);
        NodeCacheListener nodeCacheListener = new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("监听到数据库配置信息节点内容变化为：===>>>" +
                        new String(nodeCache.getCurrentData().getData()));

                String newNodeData = new String(nodeCache.getCurrentData().getData());
                Map map = (Map)JSON.parse(newNodeData);
                // 创建新的连接池
                DruidDataSource dataSource = (DruidDataSource)DruidDataSourceFactory.createDataSource(map);
                DruidUtils.setDataSource(dataSource);
                // 销毁旧的连接池
//                DruidUtils.getDataSource().close();

                // 测试数据库连接
                UserController.list();
            }
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        try {
            nodeCache.start();
        } catch (Exception e) {
            e.getMessage();
        }
    }

    /**
     * 注册监听指定节点的子节点列表
     * @param zkClient 客服端对象
     * @param serviceName 服务对象接口名 eg:com.fishleap.service.IUserService
     */
    private static void registerWatcher(CuratorFramework zkClient, String serviceName) {
        String servicePath = "/" + serviceName;
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                List<String> serviceAddresses = curatorFramework.getChildren().forPath(servicePath);
                log.info("监听到 {} 节点的子节点列表变化为 {}", servicePath, serviceAddresses);
                System.out.println("监听到 "+servicePath+" 节点的子节点列表变化为 "+serviceAddresses);
                serviceAddressMap.put(serviceName, serviceAddresses);
            }
        };
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        try {
            pathChildrenCache.start();
        } catch (Exception e) {
            e.getMessage();
        }
    }


}
